import pandas as pd
import pymysql
import time
from datetime import datetime, timedelta
from dateutil.relativedelta import relativedelta
import urllib


# 设置panda显示参数
pd.set_option("display.max_rows", 20)
pd.set_option("display.max_columns", 20)
pd.set_option("display.width", 220)

# mysql连接信息
mysql_config = {
    'host': 'rm-wz92r8qt89c31fn40oo.mysql.rds.aliyuncs.com',
    'port': 3306,
    'user': 'bi_rw',
    'password': 'Xjz@&283*-)',
}


class DatabaseConnector:
    def __init__(self, db_type, config):
        self.db_type = db_type
        self.config = config
        if db_type == 'presto':
            self.conn = dbapi.connect(**config)
        elif db_type == 'hive':
            self.conn = hive.connect(**config)
        elif db_type == 'mysql':
            self.conn = pymysql.connect(**config)
        self.cursor = self.conn.cursor()

    def ddl(self, cmd):
        """
        执行ddl语句,
        :param cmd: ddl操作语句
        :return: None
        """
        self.cursor.execute(cmd)
        self.conn.commit()

        # 使用presto的时候, 必须要加上fetchall()之后ddl语句才会执行, 不知道为什么
        if self.db_type == 'presto':
            self.cursor.fetchall()

        print(f"Finish running ddl: {cmd}")

    def dml(self, cmd):
        """
        执行增删改操作
            hive:   操作没有返回集, 使用fetchall会报错, 影响的行数也不可见
            presto: 结果集的类型是rows, insert语句的值是插入的行数, delete操作的值是None
        :param cmd: dml操作语句
        :return: None
        """
        self.cursor.execute(cmd)
        self.conn.commit()
        if self.db_type == 'presto':
            infected_rows = self.cursor.fetchall()[0][0]
            print(f"Finish running dml ({infected_rows} rows infected): {cmd}")
        else:
            print(f"Finish running dml: {cmd}")

    def dql(self, cmd):
        """
        执行查询操作, 需要返回查询结果集的dataframe
        :param cmd: dql操作sql
        :return: dql结果
        """
        self.cursor.execute(cmd)
        dql_desc = self.cursor.description
        dql_rows = self.cursor.fetchall()
        print(f"Finish running dql: {cmd}")
        dql_df = None
        if len(dql_rows) != 0:
            columns = [col[0] for col in dql_desc]
            result = [dict(zip(columns, row)) for row in dql_rows]
            dql_df = pd.DataFrame(result)
            dql_df.columns = columns

        return dql_df


mysql_env = DatabaseConnector('mysql', mysql_config)



run_sql = """
insert into  bi_all_in_one.tmp_benefit_pay_record_6month_after_20250707
with new_benefit_pay_record as 
(
select  t1.user_id                          as user_id       
       ,t1.source                           as source        
       ,t1.agency_code                      as agency_code   
       ,t1.pay_amount                       as pay_amount    
       ,t1.apply_time                       as first_apply_time               
       ,t1.pay_status                       as first_pay_status               
       ,t1.biz_resp_code                    as first_biz_resp_code            
       ,t1.biz_resp_msg                     as first_biz_resp_msg             
       ,t1.success_time                     as first_success_time             
       ,t1.pay_channel                      as first_pay_channel              
       ,t1.share_status                     as first_share_status             
       ,t1.route_id                         as first_route_id                 
       ,t1.supplement_deduct_flag           as first_supplement_deduct_flag   
       ,t1.created_time                     as first_created_time      
       ,t2.apply_time                       as apply_time               
       ,t2.pay_status                       as pay_status               
       ,t2.biz_resp_code                    as biz_resp_code            
       ,t2.biz_resp_msg                     as biz_resp_msg             
       ,t2.success_time                     as success_time             
       ,t2.pay_channel                      as pay_channel              
       ,t2.share_status                     as share_status             
       ,t2.route_id                         as route_id                 
       ,t2.supplement_deduct_flag           as supplement_deduct_flag   
       ,t2.created_time                     as created_time   
       ,t3.partner_order_no 
       ,t3.refund_time    
       ,t3.refund_amount 
       ,t3.full_refund 
       ,t3.agreement_refund 
       ,t3.refund_reason 
       ,t3.own_merchant_id 
       ,t3.own_refund_amt 
       ,t3.supplier_merchant_id 
       ,t3.supplier_refund_amt 
       ,t3.order_status 
from membership_benefit.benefit_pay_record t1 
left join membership_benefit.benefit_pay_record t2  
       on t1.user_id = t2.user_id 
      and t1.source = t2.source 
      and t1.pay_amount = t2.pay_amount 
      and t1.agency_code = t2.agency_code 
left join membership_benefit.refund_order t3 on t2.user_id = t3.user_id and t2.order_no = t3.partner_order_no and t2.agency_code = t3.agency_code 
where t1.supplement_deduct_flag = 0  -- 首扣标志
  and substr(t1.apply_time,1,7) >= '2025-06'
)-- select sum(distinct(if(pay_status = 'Y' , pay_amount ,0))) from new_benefit_pay_record;
,
first_pay_info as 
(
select 
      date(apply_time) as first_pay_date
     ,source
     ,agency_code
     ,count(distinct user_id) as first_pay_ppl
     ,count(distinct(if(pay_status = 'Y',user_id ,null)))    as first_pay_succ_ppl
     ,sum(if(pay_status = 'Y',pay_amount,0))                 as first_pay_succ_amt
     ,'{check_date}'                                        as check_date
from membership_benefit.benefit_pay_record t1 
where t1.supplement_deduct_flag = 0  -- 首扣标志
  and date(apply_time) <= date '{check_date}'    
  and substr(apply_time,1,7) >= '2025-06'   
group by 1,2,3 ,'{check_date}' 
)-- select  * from first_pay_info;
,
other_pay_info as  -- 补扣数据去重，由于关联退款信息，一条补扣数据会对应多笔退款数据
(
select  
      date(first_apply_time)  as first_pay_date
     ,source
     ,agency_code
     ,count(distinct user_id)                                as pay_ppl -- 补扣人数
     ,count(distinct(if(pay_status = 'Y',user_id ,null)))    as pay_succ_ppl -- 补扣成功人数
     ,sum(if(pay_status = 'Y' , pay_amount ,0 ))             as pay_succ_amt -- 补扣成功金额
     ,'{check_date}'     as check_date 
 from (
 select  
      distinct user_id       
      ,source        
      ,agency_code   
      ,pay_amount    
      ,first_apply_time               
      ,first_pay_status               
      ,first_biz_resp_code            
      ,first_biz_resp_msg             
      ,first_success_time             
      ,first_pay_channel              
      ,first_share_status             
      ,first_route_id                 
      ,first_supplement_deduct_flag   
      ,first_created_time      
      ,apply_time               
      ,pay_status               
      ,biz_resp_code            
      ,biz_resp_msg             
      ,success_time             
      ,pay_channel              
      ,share_status             
      ,route_id                 
      ,supplement_deduct_flag   
      ,created_time   
from new_benefit_pay_record) t 
 where date(apply_time) <= date '{check_date}'   
   and first_apply_time != apply_time -- 剔除首扣客户
group by 1,2,3 ,'{check_date}' 
)-- select * from other_pay_info;
,
refund_info as 
(
select  
      date(first_apply_time)  as first_pay_date
     ,source
     ,agency_code
     ,count(distinct(if(partner_order_no is not null,user_id,null)))                            as refund_ppl -- 退款人数
     ,count(distinct(if(partner_order_no is not null and order_status =  7 ,user_id ,null)))    as refund_succ_ppl -- 退款成功人数
     ,sum(if(partner_order_no is not null and order_status =  7, refund_amount ,0 ))            as refund_succ_amt -- 退款成功金额
     , '{check_date}'    as check_date 
from new_benefit_pay_record
where date(apply_time) <= date '{check_date}'   
group by 1,2,3 , '{check_date}'   
) -- select  * from refund_info ; 
select 
      t1.first_pay_date                  as first_pay_date               -- as "首扣日期"
     ,t1.source                          as source                       -- as "渠道号"
     ,t4.name      		                 as name      		             -- as "渠道名称"
     ,t1.agency_code                     as agency_code                  -- as "机构编码"
     ,t3.merchant_name                   as merchant_name                -- as "机构名称"
     ,t1.first_pay_ppl                   as first_pay_ppl                -- as "首扣人数"
     ,t1.first_pay_succ_ppl              as first_pay_succ_ppl           -- as "首扣成功人数"
     ,t1.first_pay_succ_amt              as first_pay_succ_amt           -- as "首扣成功金额"
     ,coalesce(t2.pay_ppl      , 0)      as pay_ppl                      -- as "补扣人数"
     ,coalesce(t2.pay_succ_ppl , 0)      as pay_succ_ppl                 -- as "补扣成功人数"
     ,coalesce(t2.pay_succ_amt , 0)      as pay_succ_amt                 -- as "补扣成功金额"
     ,coalesce(refund_ppl      , 0)      as refund_ppl                   -- as "退款人数"
     ,coalesce(refund_succ_ppl , 0)      as refund_succ_ppl              -- as "退款成功人数"
     ,coalesce(refund_succ_amt , 0)      as refund_succ_amt              -- as "退款成功金额"
     ,t1.check_date                      as check_date                   -- as "统计日期"
from first_pay_info t1 
left join other_pay_info t2 on t1.first_pay_date = t2.first_pay_date and t1.source = t2.source and t1.agency_code = t2.agency_code and t1.check_date = t2.check_date
left join refund_info t5 on t1.first_pay_date = t5.first_pay_date and t1.source = t5.source and t1.agency_code = t5.agency_code and t1.check_date = t5.check_date
join membership_benefit.merchant t3 on t1.agency_code = t3.merchant_code
join membership_benefit.channel_config t4 on t1.source = t4.source
order by 1 desc ,3,2
"""

# 循环开始日期 结束日期
start_date = datetime.strptime('2025-07-07', '%Y-%m-%d') 
end_date = datetime.strptime('2025-07-08', '%Y-%m-%d')

end_date_str = end_date.strftime('%Y-%m-%d')

while start_date <= end_date:
    start_date_str = start_date.strftime('%Y-%m-%d')

    params = {
        'check_date': start_date_str,
    }
    #mysql_env.dml(run_sql.format(**params)) # 执行
    print('**********')
    print(run_sql.format(**params)) # 检查打印循环命令是否传如参数正确
    
    start_date += timedelta(days=1)

